package com.iflytek.controller;

import com.iflytek.bean.AreaControl;
import com.iflytek.bean.MonitorInfo;
import com.iflytek.bean.MyTrigger;
import com.iflytek.schema.JSONDeserializationSchema;
import com.iflytek.utils.BloomFilterUtil;
import com.iflytek.utils.JedisUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.util.HashMap;
import java.util.Properties;

public class _06VehicleDistributionV4 {



    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);  //设置并行度

        //2.设置数据源
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","hadoop102:9092");
        properties.setProperty("group.id","g5");

        FlinkKafkaConsumer<MonitorInfo> consumer = new FlinkKafkaConsumer<MonitorInfo>("topic-car",
                new JSONDeserializationSchema<>(MonitorInfo.class),properties);
        DataStreamSource<MonitorInfo> ds1 = env.addSource(consumer);
        SingleOutputStreamOperator<AreaControl> resultDf = ds1.keyBy(k -> k.getAreaId())
                .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
                .trigger(new MyTrigger())
                .apply(new WindowFunction<MonitorInfo, AreaControl, String, TimeWindow>() {

                    // 为什么要在apply前面定义一个map,而不是沿用carCount +=1
                    HashMap<String,Integer> map = new HashMap<String,Integer>();

                    @Override
                    public void apply(String key, TimeWindow window, Iterable<MonitorInfo> input, Collector<AreaControl> out) throws Exception {

                        // redis中的bitmaps 跟 布隆过滤器 有啥关系？
                        // 布隆过滤器是一种数据结构，可以理解为算法，算法落地--> bitmaps
                        Jedis jedis = JedisUtils.getJedis();
                        String start = DateFormatUtils.format(window.getStart(), "yyyy-MM-dd HH:mm");
                        String end = DateFormatUtils.format(window.getEnd(), "yyyy-MM-dd HH:mm");
                        String  redisKey = "";

                        // 这个迭代器中每次都只有一条数据，所以 input 不会出现内存溢出的情况
                        for (MonitorInfo monitorInfo : input) {
                            redisKey = "area:"+key+":"+start;
                            int[] offsets = BloomFilterUtil.getOffsets(monitorInfo.getCar());
                            boolean r1 = jedis.getbit(redisKey, offsets[0]);
                            boolean r2 = jedis.getbit(redisKey, offsets[1]);
                            if(r1 == false || r2 == false){
                                if(map.containsKey(redisKey)){
                                    map.put(redisKey,map.get(redisKey)+1);
                                }else{
                                    map.put(redisKey,1);
                                }

                            }

                            jedis.setbit(redisKey,offsets[0],true);
                            jedis.setbit(redisKey,offsets[1],true);
                        }

                        // 每一个区域 近10分钟之内有多少量车
                        out.collect(new AreaControl(null, key, map.get(redisKey), start, end));
                    }
                });
        // 修改数据进入redis ,因为
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("bigdata01").setPort(6379).setPassword("123456").build();
        resultDf.addSink(new RedisSink<>(conf, new RedisMapper<AreaControl>() {
            /**
             *    key                       value
             * 区域编号:窗口结束时间         区域内车的数量
             */
            @Override
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.SET);  //set
            }

            @Override
            public String getKeyFromData(AreaControl data) {

                return "area:"+data.getAreaId()+":"+data.getWindowEnd();
            }

            @Override
            public String getValueFromData(AreaControl data) {
                return data.getCarCount().toString();
            }
        }));

        env.execute();
    }

}
